module("tair", package.seeall)

_VERSION = '1.0'

local tair_manager = require "ngx.tair"
local cjson = require "cjson.safe"

-- global
-- some application don't want need retrun uriencode data, they
-- should be responsible for data safe, they use uri with ec=0
local need_uriencode = true
--local command_handler = {}
local rpcode = {
  [444]    = {["code"] =  444,   ["status"] = 500, ["msg"] = "nginx error"},
  [0]      = {["code"] =  0,     ["status"] = 200, ["msg"] = "success"},
  [-7]     = {["code"] = -7,     ["status"] = 400, ["msg"] = "namespace range error"},
  [-3979]  = {["code"] = -3979,  ["status"] = 400, ["msg"] = "count reach zero"},
  [-3981]  = {["code"] = -3981,  ["status"] = 400, ["msg"] = "can not override"},
  [-3982]  = {["code"] = -3982,  ["status"] = 400, ["msg"] = "invaild argument"},
  [-3983]  = {["code"] = -3983,  ["status"] = 200, ["msg"] = "partly success"},
  [-3984]  = {["code"] = -3984,  ["status"] = 500, ["msg"] = "migrate busy"},
  [-3985]  = {["code"] = -3985,  ["status"] = 500, ["msg"] = "duplicate busy"},
  [-3986]  = {["code"] = -3986,  ["status"] = 500, ["msg"] = "write not on master"},
  [-3987]  = {["code"] = -3987,  ["status"] = 500, ["msg"] = "server can not work"},
  [-3989]  = {["code"] = -3989,  ["status"] = 500, ["msg"] = "timeout"},
  [-3990]  = {["code"] = -3990,  ["status"] = 500, ["msg"] = "send packet failed"},
  [-3991]  = {["code"] = -3991,  ["status"] = 400, ["msg"] = "item size error"},
  [-3992]  = {["code"] = -3992,  ["status"] = 400, ["msg"] = "out of range"},
  [-3994]  = {["code"] = -3994,  ["status"] = 400, ["msg"] = "serialize error"},
  [-3996]  = {["code"] = -3996,  ["status"] = 400, ["msg"] = "type not match"},
  [-3997]  = {["code"] = -3997,  ["status"] = 400, ["msg"] = "version error"},
  [-3998]  = {["code"] = -3998,  ["status"] = 400, ["msg"] = "data not exist"},
  [-3999]  = {["code"] = -3999,  ["status"] = 500, ["msg"] = "server error"},
  [-20001] = {["code"] = -20001, ["status"] = 400, ["msg"] = "target is not number"},
  [-20002] = {["code"] = -20002, ["status"] = 400, ["msg"] = "target is not integer"},
  [-20003] = {["code"] = -20003, ["status"] = 400, ["msg"] = "target is not double"},
  [-20004] = {["code"] = -20004, ["status"] = 400, ["msg"] = "data already exist"},
  [-20005] = {["code"] = -20005, ["status"] = 400, ["msg"] = "range have none"},
  [-20006] = {["code"] = -20006, ["status"] = 400, ["msg"] = "data overflow"},
  [-20007] = {["code"] = -20007, ["status"] = 400, ["msg"] = "data len too long"},
  [-20008] = {["code"] = -20008, ["status"] = 400, ["msg"] = "storage engine not support"},
}

-- ["command"] = {
---- ["method"] = "GET|POST|DELETE",
---- ["op"] = "command",
---- ["belongs"] = "keys|list|hash|sorted_set|set|common",
---- ["rw"] = "r|w",
-- }
local tair_command = {
  ["get"]              = {["method"] = "GET",    ["op"] = "get",              ["belong"] = "keys",       ["rw"] = "r"},
  ["put"]              = {["method"] = "POST",   ["op"] = "put",              ["belong"] = "keys",       ["rw"] = "w"},
  ["getset"]           = {["method"] = "POST",   ["op"] = "getset",           ["belong"] = "keys",       ["rw"] = "w"},
  ["putnx"]            = {["method"] = "POST",   ["op"] = "putnx",            ["belong"] = "keys",       ["rw"] = "w"},
  ["incr"]             = {["method"] = "POST",   ["op"] = "incr",             ["belong"] = "keys",       ["rw"] = "w"},
  ["numericplus"]      = {["method"] = "POST",   ["op"] = "numericplus",      ["belong"] = "keys",       ["rw"] = "w"},
  ["delete"]           = {["method"] = "DELETE", ["op"] = "delete",           ["belong"] = "common",     ["rw"] = "w"},
  ["exists"]           = {["method"] = "GET",    ["op"] = "exists",           ["belong"] = "common",     ["rw"] = "r"},
  ["flusharea"]        = {["method"] = "DELETE", ["op"] = "flusharea",        ["belong"] = "common",     ["rw"] = "w"},
  ["hgetall"]          = {["method"] = "GET",    ["op"] = "hgetall",          ["belong"] = "hash",       ["rw"] = "r"},
  ["hmget"]            = {["method"] = "GET",    ["op"] = "hmget",            ["belong"] = "hash",       ["rw"] = "r"},
  ["hmset"]            = {["method"] = "POST",   ["op"] = "hmset",            ["belong"] = "hash",       ["rw"] = "w"},
  ["hdel"]             = {["method"] = "DELETE", ["op"] = "hdel",             ["belong"] = "hash",       ["rw"] = "w"},
  ["hincrby"]          = {["method"] = "POST",   ["op"] = "hincrby",          ["belong"] = "hash",       ["rw"] = "w"},
  ["hkeys"]            = {["method"] = "GET",    ["op"] = "hkeys",            ["belong"] = "hash",       ["rw"] = "r"},
  ["hvals"]            = {["method"] = "GET",    ["op"] = "hvals",            ["belong"] = "hash",       ["rw"] = "r"},
  ["lpush"]            = {["method"] = "POST",   ["op"] = "lpush",            ["belong"] = "list",       ["rw"] = "w"},
  ["rpush"]            = {["method"] = "POST",   ["op"] = "rpush",            ["belong"] = "list",       ["rw"] = "w"},
  ["lpop"]             = {["method"] = "POST",   ["op"] = "lpop",             ["belong"] = "list",       ["rw"] = "w"},
  ["rpop"]             = {["method"] = "POST",   ["op"] = "rpop",             ["belong"] = "list",       ["rw"] = "w"},
  ["lrange"]           = {["method"] = "GET",    ["op"] = "lrange",           ["belong"] = "list",       ["rw"] = "r"},
  ["llen"]             = {["method"] = "GET",    ["op"] = "llen",             ["belong"] = "list",       ["rw"] = "r"},
  ["lrem"]             = {["method"] = "DELETE", ["op"] = "lrem",             ["belong"] = "list",       ["rw"] = "w"},
  ["ltrim"]            = {["method"] = "POST",   ["op"] = "ltrim",            ["belong"] = "list",       ["rw"] = "w"},
  ["zrange"]           = {["method"] = "GET",    ["op"] = "zrange",           ["belong"] = "sorted_set", ["rw"] = "r"},
  ["zrevrange"]        = {["method"] = "GET",    ["op"] = "zrevrange",        ["belong"] = "sorted_set", ["rw"] = "r"},
  ["zadd"]             = {["method"] = "POST",   ["op"] = "zadd",             ["belong"] = "sorted_set", ["rw"] = "r"},
  ["zrem"]             = {["method"] = "DELETE", ["op"] = "zrem",             ["belong"] = "sorted_set", ["rw"] = "r"},
  ["zscore"]           = {["method"] = "GET",    ["op"] = "zscore",           ["belong"] = "sorted_set", ["rw"] = "r"},
  ["zcount"]           = {["method"] = "GET",    ["op"] = "zcount",           ["belong"] = "sorted_set", ["rw"] = "r"},
  ["zremrangebyscore"] = {["method"] = "DELETE", ["op"] = "zremrangebyscore", ["belong"] = "sorted_set", ["rw"] = "w"},
  ["zcard"]            = {["method"] = "GET",    ["op"] = "zcard",            ["belong"] = "sorted_set", ["rw"] = "w"},
  ["sadd"]             = {["method"] = "POST",   ["op"] = "sadd",             ["belong"] = "set",        ["rw"] = "w"},
  ["srem"]             = {["method"] = "DELETE", ["op"] = "srem",             ["belong"] = "set",        ["rw"] = "w"},
  ["smembers"]         = {["method"] = "GET",    ["op"] = "smembers",         ["belong"] = "set",        ["rw"] = "r"},
  ["scard"]            = {["method"] = "GET",    ["op"] = "scard",            ["belong"] = "set",        ["rw"] = "r"},
  ["getrange"]         = {["method"] = "GET",    ["op"] = "getrange",         ["belong"] = "prefix",     ["rw"] = "r"},
  ["prefixget"]        = {["method"] = "GET",    ["op"] = "prefixget",        ["belong"] = "prefix",     ["rw"] = "r"},
  ["prefixput"]        = {["method"] = "POST",   ["op"] = "prefixput",        ["belong"] = "prefix",     ["rw"] = "w"},
  ["prefixremove"]     = {["method"] = "DELETE", ["op"] = "prefixremove",     ["belong"] = "prefix",     ["rw"] = "w"}
}

local function table_url_encode(data)
  if type(data) ~= "table" then
    return false
  end

  if need_uriencode == false then
    return true
  end

  for k,v in pairs(data) do
    if type(v) == "table" then
      local ok = table_url_encode(v)
      if not ok or ok == false then
        return false
      end
    elseif type(v) == "string" then
      if k == "key" or k == "value" or k == "field" then
        local encode_v = ngx.escape_uri(v)
        data[k] = encode_v
      end
    end
  end
  return true
end

local function table_url_decode(data)
  if type(data) ~= "table" then
    return false
  end

  for k,v in pairs(data) do
    if type(v) == "table" then
      local ok = table_url_decode(v)
      if not ok or ok == false then
        return false
      end
    elseif type(v) == "string" then
      if k == "key" or k == "value" or k == "field" then
        local decode_v = ngx.unescape_uri(v)
        data[k] = decode_v
      end
    end
  end
  return true
end

local function response_err(status, code, msg)
  ngx.status = status
  ngx.print(cjson.encode({["code"] = code, ["msg"] = msg}))
  ngx.exit(ngx.OK)
  return NGX_ERROR
end

local function match_group(ip, pattern_table)
  if not pattern_table or not ip then
    return nil
  end

  for pattern,group in pairs(pattern_table) do
    if string.find(ip, pattern) then
      group = group:match("^%s*(.-)%s*$")
      return group
    end
  end

  return nil
end

-- return {["for_read"] = {sorted list}, ["for_write"] = {sorted list}}
local function sort_by_read_write(config)
  if type(config) ~= "table" then
    return nil
  end

  local clusters = config.clusters
  if not clusters then
    return nil
  end

  local for_read = {}
  local for_write = {}
  for k,v in ipairs(clusters) do
    if type(v) == "table" then
      for_write[k] = v
      for_read[k] = v
--      local read_ele = {}
--      for a,b in pairs(v) do
--        read_ele[a] = b
--      end
--      for_read[k] = read_ele
    end
  end

  table.sort(for_read, function(a, b)
    local a_read_weight = tonumber(a.readWeight)
    local b_read_weight = tonumber(b.readWeight)
    if not a_read_weight then
      return false
    elseif not b_read_weight then
      return true
    else
      return a_read_weight > b_read_weight
    end
  end)

  table.sort(for_write, function(a, b)
    local a_write_weight = tonumber(a.writeWeight)
    local b_write_weight = tonumber(b.writeWeight)
    if not a_write_weight then
      return false
    elseif not b_write_weight then
      return true
    else
      return a_write_weight > b_write_weight
    end
  end)

  local sum_weight = 0
  for k,v in ipairs(for_read) do
    if type(v) == "table" then
      local read_weight = tonumber(v.readWeight)
      if not read_weight then
        return nil
      end
      sum_weight = sum_weight + read_weight
      v.readWeight = sum_weight
    end
  end
  for_read.sum_weight = sum_weight

  return {["for_read"] = for_read, ["for_write"] = for_write}
end

local function choose_configserver_for_read(group_config)
  if type(group_config) ~= "table" then
    return nil
  end

  local for_read = group_config.for_read
  if type(for_read) ~= "table" then
    return nil
  end
  local sum_weight = for_read.sum_weight
  local randInt = -1
  if sum_weight ~= 0 then
    math.randomseed(os.time())
    randInt = math.random(0, sum_weight - 1)
  end

  local will_read = nil
  for k,v in ipairs(for_read) do
    if type(v) == "table" then
      local read_weight = tonumber(v.readWeight)
      if not read_weight then
        return nil
      end
      will_read = v
      if read_weight > randInt then
        return will_read.address
      end
    end
  end
  return will_read.address
end

local function choose_configserver_for_write(group_config)
  if type(group_config) ~= "table" then
    return nil
  end

  local for_write = group_config.for_write
  if type(for_write) ~= "table" then
    return nil
  end
  local will_write = {}
  for k,v in ipairs(for_write) do
    if type(v) == "table" then
      local write_weight = tonumber(v.writeWeight)
      if not write_weight or write_weight <= 0 then
        return will_write
      end
      will_write[k] = v.address
    end
  end

  return will_write
end

-- data must be table or nil
local function response_msg(data, code, msg)
  local respcode = {}
  if not code then
    respcode = rpcode[444]
    if not msg then
      return response_err(respcode.status, respcode.code, respcode.msg)
    else
      return response_err(respcode.status, respcode.code, msg)
    end
  end

  respcode = rpcode[code]
  if not respcode then
    return response_err(500, code, "unknown tair response code")
  end

  ngx.status = respcode.status
  if type(data) == "table" then
    data.code = code
    if not msg then
      data.msg = respcode.msg
    else
      data.msg = msg
    end

    local ok = table_url_encode(data)
    if not ok or ok == false then
      return response_err(respcode.status, -3994, "url encode error")
    end

    ngx.print(cjson.encode(data))
  elseif type(data) == "nil" then
    local tmp = {}
    tmp.code = code
    if not msg then
      tmp.msg = respcode.msg
    else
      tmp.msg = msg
    end
    ngx.print(cjson.encode(tmp))
  end

  if respcode.status ~=200 then
    return NGX_ERROR
  else
    return NGX_OK
  end
end


-- tair init

local function tair_init(groupname)
  local tair, msg = tair_manager:init(groupname);
  if not tair then
    ngx.log(ngx.ERR, "groupname: " .. groupname .. " " .. msg)
    return nil
  end
  return tair
end

-- command handler impl
function tair_command.hincrby.body_parse(res)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()
  if not body or string.len(body) == 0 then
    return response_msg(nil, -3982, "body too large or zero body")
  end

  local body = cjson.decode(body)
  if not body then
    return response_msg(nil, -3982, "decode body error")
  end

  -- check body table
  -- example: {"key": "k", "field": "f", "count": 123, "version": 0, "expire": 12}
  if type(body) ~= "table" then
    return response_msg(nil, -3982, "decode body error")
  end

  local key = body.key
  if not key or type(key) ~= "string" then
    return response_msg(nil, -3982, "key type error")
  end

  local field = body.field
  if not field or type(field) ~= "string" then
    return response_msg(nil, -3982, "field type error")
  end

  local count = body.count
  if not count or type(count) ~= "number" then
    return response_msg(nil, -3982, "count is not exist")
  end

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/hash/namespace?param
  --         1    2      3

  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.hincrby.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return  tair:hincrby(body.area, body)
  -- respdata {["count"] = 123}
  -- resp msg {"count": 123, "code": 0, "msg": "success"}
end

function tair_command.hdel.body_parse(res)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()

  if body and string.len(body) > 0 then
    body = cjson.decode(body)
    if not body then
      return response_msg(nil, -3982, "decode body error")
    end
  end

  -- check body table
  -- example: {"version": 0, "expire": 12} or {}
  if not body then
    body = {}
  elseif type(body) ~= "table" then
    return response_msg(nil, -3982, "decode body error")
  end

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/hash/namespace/key/field?param
  --         1    2      3      4    5
  if not res[4] or type(res[4]) ~= "string" then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if not res[5] or type(res[5]) ~= "string" then
    return response_msg(nil, -3982, "get body error, field is not exist")
  end
  body.field = res[5]

  if res[6] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype;
  end

  if type(ngx.var.arg_ftype) == "string" then
    body.ftype = ngx.var.arg_ftype;
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.hdel.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:hdel(body.area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.hgetall.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/hash/namespace/key?param
  --         1    2      3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype;
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.hgetall.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:hgetall(body.area, body)
  -- respdata {["data"] = {
  --                    {["field"] = "f1", ["value"] = "v1"},
  --                    {["field"] = "f2", ["value"] = "v2"}
  --                    },
  --           ["version"] = 0
  --           }
  --
  -- response msg {"data":[
  --                    {"field": "f1", "value": "v1"},
  --                    {"field": "f2", "value": "v2"}
  --                    ],
  --               "version": 0,
  --               "code": 0,
  --               "msg": "success"
  --              }
end

local function key_fields_get_body(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/hash/namespace/key/field1,field2,field3?param
  --         1    2       3      4        5
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if not res[5] then
    return response_msg(nil, -3982, "get body error, fields is not exist")
  end

  if type(res[5]) ~= "table" then
    local fields = {}
    fields[1] = tostring(res[5])
    if not fields[1] then
      return response_msg(nil, -3982, "get body error, fields is not exist")
    end
    body.fields = fields
  else
    body.fields = res[5]
  end

  if res[6] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  if type(ngx.var.arg_ptype) == "string" then
    body.ktype = ngx.var.arg_ptype
  end

  if type(ngx.var.arg_ftype) == "string" then
    body.ftype = ngx.var.arg_ftype
  end

  if type(ngx.var.arg_stype) == "string" then
    body.ftype = ngx.var.arg_stype
  end

  local tfields = {}
  if not body.ftype then
    for k,v in pairs(body.fields) do
      local field = {["field"] = v}
      tfields[k] = field
    end
  else
    for k,v in pairs(body.fields) do
      local field = {["field"] = v, ["ftype"] = body.ftype}
      tfields[k] = field
    end
  end
  body.data = tfields

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.hmget.body_parse(res)
  -- /tair/group/prefix/namespace/key/field1,field2,field3?param
  --         1    2       3      4        5
  return key_fields_get_body(res)
end

function tair_command.hmget.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:hmget(body.area, body)
  -- respdata {["data"] = {
  --                    {["field"] = "f1", ["value"] = "v1"},
  --                    {["field"] = "f2", ["value"] = "v2"}
  --                    {["field"] = "f3", ["value"] = nil }
  --                    },
  --           ["version"] = 0
  --           }
  --
  -- response msg {"data":[
  --                    {"field": "f1", "value": "v1"},
  --                    {"field": "f2", "value": "v2"}
  --                    {"field": "f3", "value": nil}
  --                    ],
  --               "version": 0,
  --               "code": 0,
  --               "msg": "success"
  --              }
end

function tair_command.hmset.body_parse(res)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()

  if body and string.len(body) > 0 then
    body = cjson.decode(body)
    if not body then
      return response_msg(nil, -3982, "decode body error")
    end
  end

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/hash/namespace?param
  --        1     2     3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end
  -- {
  -- "key": "k",
  -- "data":[
  --   {"field": "f1", "value": "v1"},
  --   {"field": "f2", "value": "v2"}
  --   ],
  --   "version": "0",
  --   "expire": "-1"
  -- }

  if not body.data or type(body.data) ~= "table" or not next(body.data) then
    return response_msg(nil, -3982, "get body error, body is not exist")
  end

  if not body.key or type(body.key) ~= "string" then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.hmset.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:hmset(body.area, body)
  -- error:
  --      respdata {["data"] = {
  --                    {["field"] = "f1", ["value"] = "v1"},
  --                    {["field"] = "f2", ["value"] = "v2"}
  --                    },
  --           }
  -- or success:
  --      respdata {}
  --
  -- error:
  --      response msg {"data":[
  --                    {"field": "f1", "value": "v1"},
  --                    {"field": "f2", "value": "v2"}
  --                    ],
  --               "code": errcode, (for example -20007, data reach max length)
  --               "msg": "errmsg"
  --              }
  -- or success:
  --      response msg {"code": errcode, "msg": "errmsg"}
  --
end

function tair_command.prefixput.body_parse(res)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()

  if body and string.len(body) > 0 then
    body = cjson.decode(body)
    if not body then
      return response_msg(nil, -3982, "decode body error")
    end
  end

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/prefix/namespace?param
  --        1     2     3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end
  -- example: {"pkey": "p", "skey": "s", "value": "v", "version": 0, "expire": 12}
  -- example: {"pkey": "p", "skey": "s", "value": "v", "expire": 12}
  -- example: {"pkey": "p", "skey": "s", "value": "v"}

  if not body.pkey or type(body.pkey) ~= "string" then
    return response_msg(nil, -3982, "get body error, pkey is not exist")
  end

  if not body.skey or type(body.skey) ~= "string" then
    return response_msg(nil, -3982, "get body error, skey is not exist")
  end

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.prefixput.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:prefixput(body.area, body)
  -- success:
  --      response msg {"code": 0, "msg": "success"}
  --
end

local function lrpush_body_parse(res,lr)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()

  if body and string.len(body) > 0 then
    body = cjson.decode(body)
    if not body then
      return response_msg(nil, -3982, "decode body error")
    end
  end

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/list/namespace?param
  --        1     2     3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end
  -- {
  -- "key": "k",
  -- "data":[//start index is 0
  --   {"value": "v1", "index": 1},
  --   {"value": "v2", "index": 2},
  --   {"value": "v3", "index": 3}
  --   ],
  --   "version": 0,
  --   "expire": -1
  -- }

  if not body.data or type(body.data) ~= "table" or not next(body.data) then
    return response_msg(nil, -3982, "get body error, body is not exist")
  end

  local data = body.data
  for k,v in pairs(data) do
    if type(v) ~= "table" then
      return response_msg(nil, -3982, "data format error")
    end
    if type(v.index) ~= "number" then
      return response_msg(nil, -3982, "index is not number")
    end
  end

  if not body.key or type(body.key) ~= "string" then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  -- left 0, right 1
  body.lr = lr or 0

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

local function lrpush_execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:lrpush(body.area, body)
  -- error:
  --    respdata {["data"] = {
  --                    {["value"] = "v1", "index" = 0},
  --                    {["value"] = "v2", "index" = 1}
  --                    },
  --           }
  -- or success:
  --    respdata {}
  --
  --error:
  --    response msg {"data":[
  --                    {"value": "v1", "index": 0},
  --                    {"value": "v2", "index": 1}
  --                    ],
  --               "code": errcode, (for example -20007, data reach max length)
  --               "msg": "errmsg"
  --              }
  --or success:
  --    response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.lpush.body_parse(res)
  return lrpush_body_parse(res, 0)
end

function tair_command.lpush.execute(body)
  return lrpush_execute(body)
end

function tair_command.rpush.body_parse(res)
  return lrpush_body_parse(res, 1)
end

function tair_command.rpush.execute(body)
  return lrpush_execute(body)
end

local function lrpop_body_parse(res, lr)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()

  if body and string.len(body) > 0 then
    body = cjson.decode(body)
    if not body then
      return response_msg(nil, -3982, "decode body error")
    end
  end

  -- check body table
  -- example: {"key": "k", "count": 1, "version": 0, "expire": 12} or {}
  if not body or type(body) ~= "table" then
    body = {}
  end

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  local key = body.key
  if not key or type(key) ~= "string" then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end

  local count = body.count
  if not count or type(count) ~= "number" then
    return response_msg(nil, -3982, "get body error, count is not exist")
  end

  -- /tair/group/list/namespace?param
  --         1    2      3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end

  -- left is 0, right is 1
  body.lr = lr or 0

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

local function lrpop_execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:lrpop(body.area, body)
  --    respdata {["data"] = { start index with 0
  --                    {["value"] = "v1", "index" = 0},
  --                    {["value"] = "v2", "index" = 1}
  --                    },
  --               ["version"] = 0
  --           }
  --
  --    response msg {"data":[
  --                    {"value": "v1", "index": 0},
  --                    {"value": "v2", "index": 1}
  --                    ],
  --               "version": 0,
  --               "code": 0
  --               "msg": "success"
  --              }
  --
end

function tair_command.lpop.body_parse(res)
  return lrpop_body_parseody(res, 0)
end

function tair_command.lpop.execute(body)
  return lrpop_execute(body)
end

function tair_command.rpop.body_parse(res)
  return lrpop_body_parse(res, 1)
end

function tair_command.rpop.execute(body)
  return lrpop_execute(body)
end

function tair_command.zadd.body_parse(res)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()
  if not body or string.len(body) == 0 then
    return response_msg(nil, -3982, "body too large or zero body")
  end

  local body = cjson.decode(body)
  if not body then
    return response_msg(nil, -3982, "decode body error")
  end

  -- check body table
  -- example: {"key":"k", "value":"v", "score": 123, "version": 0, "expire": 12}
  if type(body) ~= "table" then
    return response_msg(nil, -3982, "decode body error")
  end

  if not body.score or type(body.score) ~= "number" then
    return response_msg(nil, -3982, "score is not exist")
  end
  body.score = body.score or 0

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  if not body.key or type(body.key) ~= "string" then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end

  if not body.value or type(body.value) ~= "string" then
    return response_msg(nil, -3982, "get body error, value is not exist")
  end

  -- /tair/group/sorted_set/namespace?param
  --         1         2      3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.zadd.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:zadd(body.area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.getrange.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/prefix/namespace/prefixkey?st=xxx&ed=yyy&cmd=t&param
  --        1     2       3          4
  -- cmd:
  -- 1 range key and value
  -- 2 range value only
  -- 3 range key only
  -- 4 range key and value reverse
  -- 5 range value only reverse
  -- 6 range key only reverse

  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  if type(ngx.var.arg_ptype) == "string" then
    body.ktype = ngx.var.arg_ptype
  end

  if type(ngx.var.arg_ftype) == "string" then
    body.ftype = ngx.var.arg_ftype
  end

  if type(ngx.var.arg_stype) == "string" then
    body.ftype = ngx.var.arg_stype
  end

  local cmd = tonumber(ngx.var.arg_cmd)
  if not cmd then
    return response_msg(nil, -3982, "cmd is not exist")
  end
  body.cmd = cmd

  if type(ngx.var.arg_st) == "string" then
    body.st = ngx.var.arg_st
  end

  if type(ngx.var.arg_ed) == "string" then
    body.ed = ngx.var.arg_ed
  end

  local offset = tonumber(ngx.var.arg_offset)
  body.offset = offset or 0

  local limit = tonumber(ngx.var.arg_limit)
  body.limit = limit or 1000

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.getrange.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:getrange(body.area, body)
  --
  -- respdata {["data"] = {
  --            {["field"] = "f1", ["value"] = "v1"},
  --            {["field"] = "f2", ["value"] = "v2"},
  --            {["field"] = "f3", ["value"] = 3, ["vtype"] = "int"}
  --            },
  --            flag = "has_next" or "has_no_next"
  --          }
  --
  -- response msg {"data": [
  --                    {"field": "f1", "value": "v1"},
  --                    {"field": "f2", "value": "v2"}
  --                    {"field": "f3", "value": 3, "vtype": "int"}
  --                  ],
  --             "flag": "has_next" or "has_no_next"
  --             "code": 0,
  --             "msg": "msg"
  --            }
  --
end

function tair_command.prefixremove.body_parse(res)
  -- /tair/group/hash/namespace/key/field1,field2,field3?param
  --         1    2       3      4        5
  return key_fields_get_body(res)
end

function tair_command.prefixremove.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:prefixremove(body.area, body)
  -- response msg {"code": 0, "msg": "success"}
end

function tair_command.prefixget.body_parse(res)
  -- /tair/group/hash/namespace/key/field1,field2,field3?param
  --         1    2       3      4        5
  return key_fields_get_body(res)
end

function tair_command.prefixget.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:prefixget(body.area, body)
  --
  -- respdata {["data"] = {
  --              {["field"] = "f1", ["value"] = "v1", ["version"] = 1, ["expire"] = 10, ["code"] = 0},
  --              {["field"] = "f2", ["value"] = "v2",["version"] = 1, ["expire"] = 0, ["code"] = 0},
  --              {["field"] = "f3", ["value"] = 3, ["vtype"] = "int", ["version"]["expire"] = 0, ["code"] = 0},
  --              {["field"] = 7, ["ktype"] = "int", ["value"] = "hello", ["expire"] = 0, ["code"] = 0},
  --              {["field"] = "f4"}
  --            }
  --          }
  --
  -- response msg {"data": [
  --                    {"field": "f1", "value": "v1", "version": 1, "expire": 10, "code": 0},
  --                    {"field": "f2", "value": "v2", "version": 1, "expire": 0, "code": 0},
  --                    {"field": "f3", "value": 3, "vtype": "int", "version": 1, ""},
  --                    {"field": "f4"}
  --                  ],
  --             "code": 0,
  --             "msg": "msg"
  --            }
  --
end

local function zrange_get_body(res, re)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/sorted_set/namespace/key?st=x&ed=y&param
  --         1        2         3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local st = tonumber(ngx.var.arg_st)
  if not st then
    return response_msg(nil, -3982, "start is not exist")
  end
  body.st = st

  local ed = tonumber(ngx.var.arg_ed)
  if not ed then
    return response_msg(nil, -3982, "end is not exist")
  end
  body.ed = ed

  local withscore = tonumber(ngx.var.arg_withscore)
  body.withscore = withscore or 1

  -- zrange is 0, zrevrange is 1
  body.re = re or 0

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

local function zrange_execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:zrange(body.area, body)
  --
  -- respdata {["data"] = {
  --            {["value"] = "v1", ["score"] = 1},
  --            {["value"] = "v2", ["score"] = 2.3}
  --            },
  --           ["version"] = 0
  --          }
  --
  -- response msg {"data": [
  --                    {"value": "v1", "score": 1,},
  --                    {"value": "v2", "score": 2.3}
  --                  ],
  --             "version": 0,
  --             "code": 0,
  --             "msg": "msg"
  --            }
  --
end

function tair_command.zrange.body_parse(res)
  return zrange_get_body(res, 0)
end

function tair_command.zrange.execute(body)
  return zrange_execute(body)
end

function tair_command.zrevrange.body_parse(res)
  return zrange_get_body(res, 1)
end

function tair_command.zrevrange.execute(body)
  return zrange_execute(body)
end

function tair_command.exists.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/sorted_set/namespace/key?param
  --         1        2         3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.exists.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:exists(body.area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.incr.body_parse(res)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()

  if body and string.len(body) > 0 then
    body = cjson.decode(body)
    if not body then
      return response_msg(nil, -3982, "decode body error")
    end
  end

  -- check body table
  -- example: {"key": "k", "count": 12, "init": 13, "version": 0, "expire": 12}
  if type(body) ~= "table" then
    return response_msg(nil, -3982, "get body error")
  end

  if not body.count or type(body.count) ~= "number" then
    return response_msg(nil, -3982, "not found count")
  end

  if body.init and type(body.init) ~= "number" then
    return response_msg(nil, -3982, "init type error")
  end
  body.init = body.init or 0

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  if not body.key or type(body.key) ~= "string" then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end

  -- /tair/group/keys/namespace?param
  --         1    2      3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.incr.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:incr(body.area, body)
  -- respdata {["count"] = 123}
  -- resp msg {"count": 123, "code": 0, "msg": "success"}
end

function tair_command.numericplus.body_parse(res)
  return tair_command.incr.body_parse(res)
end

function tair_command.numericplus.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:numericplus(body.area, body)
  -- respdata {["count"] = 123}
  -- resp msg {"count": 123, "code": 0, "msg": "success"}
end

function tair_command.get.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/keys/namespace/key1,key2?param
  --         1    2      3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  local tkeys = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local keys = {}
  if type(tkeys) == "string" then
    keys[1] = tkeys
  elseif type(tkeys) == "table" then
    keys = tkeys
  end

  local xkeys = {}
  for k,v in pairs(keys) do
    if type(body.ktype) == "string" then
      xkeys[k] = {["key"] = v, ["ktype"] = body.ktype}
    else
      xkeys[k] = {["key"] = v}
    end
  end

  body.data = xkeys

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.get.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:mget(body.area, body)
  -- respdata {["data"] = {
  --                {["key"] = "key1", ["value"] = "v1", ["expire"] = -1, ["version"] = 0},
  --                {["key"] = "key2", ["value"] = "100", ["expire"] = -1, ["version"] = 0},
  --                {["key"] = "key3", ["value"] = nil}
  --          }
  --
  -- response msg
  -- {
  --   "data": [
  --        {"key":  "key1", "value": "v1", "expire": 0, "version": 0},
  --        {"key":  "key2", "value": "100", "expire": 0, "version": 0},
  --        {"key":  "key3", "value": null}
  --    ],
  --    "code": 0,
  --    "msg": "success"
  -- }
end

--function command_handler.get(res)
--  local body = {}
--  if not res.area then
--    return response_msg(nil, -3982, "get body error, area is not exist")
--  end
--  body.area = res.area
--
--  if not res.group then
--    return response_msg(nil, -3982, "get body error, group is not exist")
--  end
--  body.group = res.group
--
--  -- /tair/group/keys/namespace/key?param
--  --         1    2      3       4
--  if not res[4] then
--    return response_msg(nil, -3982, "get body error, key is not exist")
--  end
--  if type(res[4]) == "table" then
--    body.key = res[4][1]
--  else
--    body.key = res[4]
--  end
--
--  local tair = tair_init(body.group)
--  if not tair then
--    return response_msg(nil, 444, "tair init failed")
--  end
--
--  local respdata, rc, msg = tair:get(body.area, body)
--  -- respdata {["key"] = "key1", ["value"] = "v1", ["expire"] = -1, ["version"] = 0}}
--  --
--  -- response msg
--  -- {
--  --    "key":  "key1",
--  --    "value": "v1",
--  --    "expire": 0,
--  --    "version": 0,
--  --    "code": 0,
--  --    "msg": "success"
--  -- }
--  return response_msg(respdata, rc, msg)
--end

local function put_get_body(res, nx)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()

  if body and string.len(body) > 0 then
    body = cjson.decode(body)
    if not body then
      return response_msg(nil, -3982, "decode body error")
    end
  end

  -- check body table
  -- example: {"key": "k", "value": "v", "version": 0, "expire": 12} or {}
  if not body or type(body) ~= "table" then
    body = {}
  end

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  if not body.key or type(body.key) ~= "string" then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end

  if not body.value or type(body.value) ~= "string" then
    return response_msg(nil, -3982, "get body error, value is not exist")
  end

  -- /tair/group/keys/namespace?param
  --         1    2      3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  body.nx = nx

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

local function put_execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:put(body.area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.getset.body_parse(res, nx)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()

  if body and string.len(body) > 0 then
    body = cjson.decode(body)
    if not body then
      return response_msg(nil, -3982, "decode body error")
    end
  end

  -- check body table
  -- example: {"key": "k", "value": "v", "version": 0, "expire": 12} or {}
  if not body or type(body) ~= "table" then
    body = {}
  end

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  if not body.key or type(body.key) ~= "string" then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end

  if not body.value or type(body.value) ~= "string" then
    return response_msg(nil, -3982, "get body error, value is not exist")
  end

  -- /tair/group/keys/namespace?param
  --         1    2      3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  body.nx = nx

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.getset.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:getset(body.area, body)
  -- respdata {
  --                {["key"] = "key2", ["value"] = "100", ["expire"] = -1, ["version"] = 0},
  --          }
  --
  -- response msg
  -- {
  --    "key":  "key1",
  --    "value": "v1",
  --    "expire": 0,
  --    "version": 0,
  --    "code": 0,
  --    "msg": "success"
  -- }
end

function tair_command.put.body_parse(res)
  return put_get_body(res, 0)
end

function tair_command.put.execute(body)
  return put_execute(body)
end

function tair_command.putnx.body_parse(res)
  return put_get_body(res, 1)
end

function tair_command.putnx.execute(body)
  return put_execute(body)
end

local function mput_get_body(res, nx)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()

  if body and string.len(body) > 0 then
    body = cjson.decode(body)
    if not body then
      return response_msg(nil, -3982, "decode body error")
    end
  end

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/keys/namespace/?param
  --        1     2     3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end
  -- {
  -- "data":[
  -- {"key": "k1","value": "v1", "version": 0, "expire": -1},
  -- {"key": "k2","value": "v2", "version": 0, "expire": -1}
  -- ],
  -- }

  if not body.data or type(boyd.data) ~= "table" then
    return response_msg(nil, -3982, "get body error, body is not exist")
  end

  local xdata = {}
  local id = 1
  for index,dat in pairs(data) do
    if not dat and type(dat) == "table" then
      if not dat.key or type(dat.key) == "table" then
        return response_msg(nil, -3982, "get body error, key error")
      end

      if not dat.value or type(dat.value) == "table" then
        return response_msg(nil, -3982, "get body error, value error")
      end

      if dat.version and type(dat.version) then
        return response_msg(nil, -3982, "version type error")
      end
      dat.version = dat.version or 0

      if dat.expire and type(dat.expire) then
        return response_msg(nil, -3982, "expire type error")
      end
      dat.expire = dat.expire or -1

      xdata[id] = dat
      id = id + 1
    end
  end

  if next(xdata) == nil then
    return response_msg(nil, -3982, "get body error, data empty")
  end

  body.data = xdata
  body.nx = nx

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

local function mput_execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:mput(area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

--[[
--not public
function tair_command.mput.body_parse(res)
  return mput_get_body(res, 0)
end

function tair_command.mput.execute(body)
  return mput_execute(body)
end

function tair_command.mputnx.body_parse(res)
  return mput_get_body(res, 1)
end

function tair_command.mputnx.execute(body)
  return mput_execute(body)
end
]]--

local function mdelete_get_body(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/keys/namespace/key1,key2?param
  --         1    2      3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  local tkeys = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local keys = {}
  if type(tkeys) == "string" then
    keys[1] = tkeys
  elseif type(tkeys) == "table" then
    keys = tkeys
  end

  local xkeys = {}
  for k,v in pairs(keys) do
    xkeys[k] = {["key"] = v}
  end
  body.data = xkeys

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

local function mdelete_execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:delete(body.area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.delete.body_parse(res)
  return mdelete_get_body(res)
end

function tair_command.delete.execute(body)
  return mdelete_execute(body)
end

function tair_command.lrange.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/list/namespace/key?st=x&ed=y&param
  --         1    2      3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local st = tonumber(ngx.var.arg_st)
  if not st then
    return response_msg(nil, -3982, "start is not exist")
  end
  body.st = st

  local ed = tonumber(ngx.var.arg_ed)
  if not ed then
    return response_msg(nil, -3982, "end is not exist")
  end
  body.ed = ed

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.lrange.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:lrange(body.area, body)
  --
  -- respdata {["data"] = {
  --            {["value"] = "v1"},
  --            {["value"] = "v2"}
  --            },
  --           ["version"] = 0
  --          }
  --
  -- response msg {"data": [
  --                    {"value": "v1"},
  --                    {"value": "v2"}
  --                  ],
  --             "version": 0,
  --             "code": 0,
  --             "msg": "msg"
  --            }
  --
end

function tair_command.llen.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/list/namespace/key?param
  --         1        2         3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.llen.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:llen(body.area, body)
  --
  -- response msg {"count": 2, "code": 0, "msg": "success"}
  --
end

function tair_command.lrem.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/list/namespace/key/value?ct=1&param
  --         1    2     3       4    5
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if not res[5] then
    return response_msg(nil, -3982, "get body error, value is not exist")
  end
  body.value = res[5]

  if res[6] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.ktype) == "string" then
    body.ktype = ngx.var.ktype
  end

  if type(ngx.var.arg_vtype) == "string" then
    body.vtype = ngx.var.vtype
  end

  local ct = tonumber(ngx.var.arg_ct)
  if not ct then
    return response_msg(nil, -3982, "count is not exist")
  end
  body.count = ct

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.lrem.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:lrem(body.area, body)
  --
  -- response msg {"count": 2, "code": 0, "msg": "success"}
  --
end

function tair_command.ltrim.body_parse(res)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()
  if not body or string.len(body) == 0 then
    return response_msg(nil, -3982, "body too large or zero body")
  end

  local body = cjson.decode(body)
  if not body then
    return response_msg(nil, -3982, "decode body error")
  end

  -- check body table
  -- example: {"key":"k", "st": 0, "ed": 1, "version": 0, "expire": 12}
  if type(body) ~= "table" then
    return response_msg(nil, -3982, "decode body error")
  end

  if not body.st or type(body.st) ~= "number" then
    return response_msg(nil, -3982, "start is not exist")
  end

  if not body.ed or type(body.ed) ~= "number" then
    return response_msg(nil, -3982, "end is not exist")
  end

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  if not body.key or type(body.key) ~= "string" then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end

  -- /tair/group/list/namespace?param
  --         1    2      3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.ltrim.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:ltrim(body.area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.hkeys.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/hash/namespace/key?param
  --         1    2      3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.hkeys.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:hkeys(body.area, body)
  -- respdata {["data"] = {
  --                    {["field"] = "f1"},
  --                    {["field"] = "f2"}
  --                    },
  --           ["version"] = 0
  --           }
  --
  -- response msg {"data":[
  --                    {"field": "f1"},
  --                    {"field": "f2"}
  --                    ],
  --               "version": 0,
  --               "code": 0,
  --               "msg": "success"
  --              }
end

function tair_command.hvals.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/hash/namespace/key?param
  --         1    2      3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.hvals.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:hvals(body.area, body)
  -- respdata {["data"] = {
  --                    {["value"] = "f1"},
  --                    {["value"] = "f2"}
  --                    },
  --           ["version"] = 0
  --           }
  --
  -- response msg {"data":[
  --                    {"value": "f1"},
  --                    {"value": "f2"}
  --                    ],
  --               "version": 0,
  --               "code": 0,
  --               "msg": "success"
  --              }
end

function tair_command.zrem.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/sorted_set/namespace/key/value?param
  --         1    2           3       4    5
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if not res[5] then
    return response_msg(nil, -3982, "get body error, value is not exist")
  end
  body.value = res[5]

  if res[6] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  if type(ngx.var.arg_vtype) == "string" then
    body.vtype = ngx.var.arg_vtype
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.zrem.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:zrem(body.area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.zscore.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/sorted_set/namespace/key/value?param
  --         1         2      3       4    5
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if not res[5] then
    return response_msg(nil, -3982, "get body error, value is not exist")
  end
  body.value = res[5]

  if res[6] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  if type(ngx.var.arg_vtype) == "string" then
    body.vtype = ngx.var.arg_vtype
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.zscore.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:zscore(body.area, body)
  --
  -- response msg {"score": 1, "code": 0, "msg": "success"}
  --
end

function tair_command.zcount.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/list/namespace/key?st=x&ed=y&param
  --         1    2      3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local st = tonumber(ngx.var.arg_st)
  if not st then
    return response_msg(nil, -3982, "start is not exist")
  end
  body.st = st

  local ed = tonumber(ngx.var.arg_ed)
  if not ed then
    return response_msg(nil, -3982, "end is not exist")
  end
  body.ed = ed

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.zcount.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:zcount(body.area, body)
  --
  -- response msg {"count": 1, "code": 0, "msg": "success"}
  --
end

function tair_command.zremrangebyscore.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/list/namespace/key?st=1.0&ed=2.0&param
  --         1    2     3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local st = tonumber(ngx.var.arg_st)
  if not st then
    return response_msg(nil, -3982, "start is not exist")
  end
  body.st = st

  local ed = tonumber(ngx.var.arg_ed)
  if not ed then
    return response_msg(nil, -3982, "end is not exist")
  end
  body.ed = ed

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.zremrangebyscore.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:zremrangebyscore(body.area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.flusharea.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/common/namespace&param
  --         1    2     3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.flusharea.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:flusharea(body.area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.sadd.body_parse(res)
  ngx.req.read_body()

  local body = ngx.req.get_body_data()
  if not body or string.len(body) == 0 then
    return response_msg(nil, -3982, "body too large or zero body")
  end

  local body = cjson.decode(body)
  if not body then
    return response_msg(nil, -3982, "decode body error")
  end

  -- check body table
  -- example: {"key":"k", "value":"v", "version": 0, "expire": 12}
  if type(body) ~= "table" then
    return response_msg(nil, -3982, "decode body error")
  end

  if body.expire and type(body.expire) ~= "number" then
    return response_msg(nil, -3982, "expire type error")
  end
  body.expire = body.expire or -1

  if body.version and type(body.version) ~= "number" then
    return response_msg(nil, -3982, "version type error")
  end
  body.version = body.version or 0

  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  if not body.key or type(body.key) ~= "string" then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end

  if not body.value or type(body.value) ~= "string" then
    return response_msg(nil, -3982, "get body error, value is not exist")
  end

  -- /tair/group/sorted_set/namespace?param
  --         1         2      3
  if res[4] then
    return response_msg(nil, -3982, "url format error")
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.sadd.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:sadd(body.area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.srem.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/set/namespace/key/value?param
  --         1    2     3       4    5
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if not res[5] then
    return response_msg(nil, -3982, "get body error, value is not exist")
  end
  body.value = res[5]

  if res[6] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  if type(ngx.var.arg_vtype) == "string" then
    body.vtype = ngx.var.arg_vtype
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.srem.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:srem(body.area, body)
  --
  -- response msg {"code": 0, "msg": "success"}
  --
end

function tair_command.smembers.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/set/namespace/key?param
  --         1    2     3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.smembers.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:smembers(body.area, body)
  --
  -- respdata {["data"] = {
  --            {["value"] = "v1"},
  --            {["value"] = "v2"}
  --            },
  --           ["version"] = 0
  --          }
  --
  -- response msg {"data": [
  --                    {"value": "v1"},
  --                    {"value": "v2"}
  --                  ],
  --             "version": 0,
  --             "code": 0,
  --             "msg": "msg"
  --            }
  --
end

function tair_command.zcard.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/set/namespace/key?param
  --         1    2     3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.zcard.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:zcard(body.area, body)
  --
  -- response msg {"count": 1, "code": 0, "msg": "success"}
  --
end

function tair_command.scard.body_parse(res)
  local body = {}
  if not res.area then
    return response_msg(nil, -3982, "get body error, area is not exist")
  end
  body.area = res.area

  if not res.group then
    return response_msg(nil, -3982, "get body error, group is not exist")
  end
  body.group = res.group

  -- /tair/group/set/namespace/key?param
  --         1    2     3       4
  if not res[4] then
    return response_msg(nil, -3982, "get body error, key is not exist")
  end
  body.key = res[4]

  if res[5] then
    return response_msg(nil, -3982, "url format error")
  end

  if type(ngx.var.arg_ktype) == "string" then
    body.ktype = ngx.var.arg_ktype
  end

  local ok = table_url_decode(body)
  if not ok or ok == false then
    return response_msg(nil, -3994, "serialize error")
  end

  return body
end

function tair_command.scard.execute(body)
  local tair = tair_init(body.group)
  if not tair then
    return nil, 444, "tair init failed"
  end

  return tair:scard(body.area, body)
  --
  -- response msg {"count": 1, "code": 0, "msg": "success"}
  --
end

local function command_process(resource, taircmd, clusters)
  local method = ngx.var.request_method
  if method ~= taircmd.method then
    return response_msg(nil, -3982, "method '" .. method .. "' and op '" .. taircmd.op .. "' forbidden")
  end

  if taircmd.op == "delete" then
    if res[2] ~= "keys" and res[2] ~= "list" and res[2] ~= "hash"
      and res[2] ~= "sorted_set" and res[2] ~= "prefix" and res[2] ~= "common" then
      return response_msg(nil, -3982, "op '" .. taircmd.op .. "' is not as '" .. "keys/list/hash/sorted_set/prefix/common'")
    end
  elseif res[2] ~= taircmd.belong then
    return response_msg(nil, -3982, "op '" .. taircmd.op .. "' is not as '" .. taircmd.belong .. "'")
  end

  local body = taircmd.body_parse(resource)
  if type(body) == "table" then
    if clusters then
      local respdata, rc, msg = nil, nil, nil
      for _, nickname in ipairs(clusters) do
        body.group = nickname
        respdata, rc, msg = taircmd.execute(body)
        if not rc and rc ~= 0 and rc ~= -3983 and rc ~= -20004 and rc ~= -20007 then
          return response_msg(respdata, rc, msg)
        end
      end
      return response_msg(respdata, rc, msg)
    else
      local respdata, rc, msg = taircmd.execute(body)
      return response_msg(respdata, rc, msg)
    end
  end
end

-- parser not care which is
local function parse_url(url)
  local res = {}

  _,_,url = string.find(url, "/tair/(.*)")
  if not url then
    return res
  end

  --?
  local address = url

  local tres = {}
  --/ address
  local t = string.byte(address, -1)
  if t == 47 then
    address = string.sub(address, 1, -2)
  end

  local id = 1
  while true do
    if not address then
      break
    end
    _,_,sub,tmp = string.find(address, "([^/]+)/(.*)")
    if sub then
      tres[id] = sub
      id = id + 1
    elseif not sub then
      if address then
        tres[id] = address
      end
      break
    end
    address = tmp
  end

  res = tres

  local ares = {}
  --, address
  for k,v in pairs(res) do
    if not v then
      break
    end
    local i = string.find(v, ",")
    if not i then
      --local vv = url_decode(v)
      ares[k] = v
    else
      local part = {}
      local id = 1
      while true do
        if not v then
          break
        end
        _,_,sub,tmp = string.find(v, "([^,]+),(.*)")
        if sub then
          --local ssub = urldecode(sub)
          part[id] = sub
          id = id + 1
        elseif not sub then
          if v then
            --local vv = urldecode(v)
            part[id] = v
            id = id + 1
          end
          break
        end
        v = tmp
      end
      ares[k] = part
    end
  end

  return ares
end

-- url/tair/groupname/keys|list|hash|sorted_set/namespace/other/?param
-- so index       1          2                     3
local function ngx_tair_parse_url(uri)
  res = parse_url(uri)
  if not res or (not next(res)) then
    return nil
  end

  local group = res[1]
  if not group or type(group) ~= "string" then
    return nil
  end
  res.group = group

  local area = tonumber(res[3])
  if not area or area < 0 then
    return nil
  end
  res.area = area

  return res
end


-- main --
function run(mode)
  local uri = ngx.var.uri
  local v = ngx.var.arg_v
  v = tonumber(v)
  if not v or v ~= 1.0 then
    return response_msg(nil, -3982, "tair API version error")
  end

  local resource = ngx_tair_parse_url(uri)
  if not resource then
    i, _ = string.find(uri, '/tair/version')
    if i ~= nil then
      return response_msg(nil, 0, _VERSION)
    else
      return response_msg(nil, -3982, "invalid url")
    end
  end

  if not ngx.var.arg_op then
    return response_msg(nil, -3982, "not found op")
  end
  resource.op = ngx.var.arg_op

  if ngx.var.arg_ec == "0" then
    need_uriencode = false
  elseif ngx.var.arg_ec == "1" then
    need_uriencode = true
  end

  local clusters = nil
  local taircmd = tair_command[resource.op];
  if not taircmd then
    return response_msg(nil, -3982, "op '" .. resource.op .. "' is not support")
  end

  if mode == "diamond" then
    clusters = get_group_from_diamond_config(taircmd.rw, resource.group)
  elseif mode == "single" then
  else
    return response_msg(nil, 444, "unknown mode")
  end

  command_process(resource, taircmd, clusters)
end

local function parse_dataId_GROUP_rule(group_config)
  local config = {}
  for k,v in string.gmatch(group_config, "([^%c]+)=([^%c]+)") do
    config[k] = v
  end
  return config
end

local function generate_local_tag(diamond_group_name, group_name, master)
  return diamond_group_name .. "~" .. group_name .. "~" .. master
end

function get_group_from_diamond_config(optype, configid)
  if configid == nil then
    return old_get_group_from_diamond_config(optype)
  else
    return new_get_group_from_diamond_config(optype, configid)
  end
end

-- optype = tair_command["cmd"].rw
-- example: put -> tair_command["put"].rw
-- special: mget -> tair_command["get"].rw
-- return: groups or nil
-- groups = {[1] = local_tag1, [2] = local_tag2, ...}
function new_get_group_from_diamond_config(optype, configid)
  if not ngx.var[configid .. '_diamond_config_version'] then return nil end
  if not ngx.var[configid .. '_default_group_config'] then return nil end

  if tair_manager[configid] == nil or tair_manager[configid].diamond_config_version == nil or
    tair_manager[configid].diamond_config_version ~= ngx.var[configid .. '_diamond_config_version'] then

    starttime = ngx.now()

    rule = parse_dataId_GROUP_rule(ngx.var[configid .. '_default_group_config'])
    if not rule then return nil end

    localip = tair_manager.localip()
    if localip == nil then
      ngx.log(ngx.ERR, "can't get local ip address !!!")
      return nil
    end

    ngx.log(ngx.ERR, "Note: local ip address is " .. localip)

    diamond_group_name = match_group(localip, rule)
    if not diamond_group_name then return nil end
    diamond_group = ngx.var[diamond_group_name]
    if not diamond_group then return nil end
    diamond_group = cjson.decode(diamond_group)
    if not diamond_group then return nil end
    diamond_group = sort_by_read_write(diamond_group)
    if not diamond_group then return nil end

    ngx.log(ngx.ERR, "-------update diamond group config---------")
    if tair_manager[configid] == nil or tair_manager[configid].diamond_config_version == nil then
      ngx.log(ngx.ERR, "if each request cause updating diamond group configure, please set `lua_code_cache` to be `on`")
    end
    tair_manager[configid] = {}
    tair_manager[configid].diamond_config_version = ngx.var[configid .. '_diamond_config_version']
    tair_manager[configid].diamond_group = diamond_group
    tair_manager[configid].diamond_group_name = diamond_group_name

    -- for read
    local address = choose_configserver_for_read(tair_manager[configid].diamond_group)
    local local_read_tag = generate_local_tag(tair_manager[configid].diamond_group_name, address.group, address.master)
    tair_manager[configid].local_read_tags = { [1] = local_read_tag }
    local address_list = choose_configserver_for_write(tair_manager[configid].diamond_group)
    -- for write
    local local_tags = {}
    for index,address in ipairs(address_list) do
      local_tags[index] = generate_local_tag(tair_manager[configid].diamond_group_name, address.group, address.master)
    end
    tair_manager[configid].local_write_tags = local_tags

    endtime = ngx.now()
    ngx.log(ngx.ERR, "-------------use time " .. (endtime - starttime) .. "------------")
  end

  if optype == "r" then
    return tair_manager[configid].local_read_tags
  elseif optype == "w" then
    return tair_manager[configid].local_write_tags
  end
  return nil
end

-- optype = tair_command["cmd"].rw
-- example: put -> tair_command["put"].rw
-- special: mget -> tair_command["get"].rw
-- return: groups or nil
-- groups = {[1] = local_tag1, [2] = local_tag2, ...}
function old_get_group_from_diamond_config(optype)
  if not ngx.var.diamond_config_version then return nil end
  if not ngx.var.default_group_config then return nil end

  if tair_manager.diamond_config_version == nil or
    tair_manager.diamond_config_version ~= ngx.var.diamond_config_version then

    starttime = ngx.now()

    rule = parse_dataId_GROUP_rule(ngx.var.default_group_config)
    if not rule then return nil end

    localip = tair_manager.localip()
    if localip == nil then
      ngx.log(ngx.ERR, "can't get local ip address !!!")
      return nil
    end

    ngx.log(ngx.ERR, "Note: local ip address is " .. localip)

    diamond_group_name = match_group(localip, rule)
    if not diamond_group_name then return nil end
    diamond_group = ngx.var[diamond_group_name]
    if not diamond_group then return nil end
    diamond_group = cjson.decode(diamond_group)
    if not diamond_group then return nil end
    diamond_group = sort_by_read_write(diamond_group)
    if not diamond_group then return nil end

    ngx.log(ngx.ERR, "-------update diamond group config---------")
    if tair_manager.diamond_config_version == nil then
      ngx.log(ngx.ERR, "if each request cause updating diamond group configure, please set `lua_code_cache` to be `on`")
    end
    tair_manager.diamond_config_version = ngx.var.diamond_config_version
    tair_manager.diamond_group = diamond_group
    tair_manager.diamond_group_name = diamond_group_name

    -- for read
    local address = choose_configserver_for_read(tair_manager.diamond_group)
    local local_read_tag = generate_local_tag(tair_manager.diamond_group_name, address.group, address.master)
    tair_manager.local_read_tags = { [1] = local_read_tag }
    local address_list = choose_configserver_for_write(tair_manager.diamond_group)
    -- for write
    local local_tags = {}
    for index,address in ipairs(address_list) do
      local_tags[index] = generate_local_tag(tair_manager.diamond_group_name, address.group, address.master)
    end
    tair_manager.local_write_tags = local_tags

    endtime = ngx.now()
    ngx.log(ngx.ERR, "-------------use time " .. (endtime - starttime) .. "------------")
  end

  if optype == "r" then
    return tair_manager.local_read_tags
  elseif optype == "w" then
    return tair_manager.local_write_tags
  end
  return nil
end
